package com.atguigu.day04;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import com.atguigu.utils.UrlCountPerWindow;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

//统计的是每个url在 每个5秒钟的滚动窗口中 的被访问次数
//增量聚合函数和全窗口聚合函数结合使用进行优化
//每个窗口只需维护一个累加器，来一条数据更新完就gc回收了，不需要存储所有数据，极大优化了内存
public class Example6 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.url)
                //5秒钟的滚动窗口
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .aggregate(new CountAgg(),new WindowResult())
                .print();


        env.execute();
    }


    //CountAgg增量聚合函数
    //AggregateFunction<IN,ACC,OUT> 三个泛型分别是 输入，累加器，输出泛型
    public static class CountAgg implements AggregateFunction<ClickEvent,Long,Long>{
        //创建空累加器
        @Override
        public Long createAccumulator() {
            return 0L;
        }


        //定义输入数据和累加器的聚合规则
        @Override
        public Long add(ClickEvent clickEvent, Long accumulator) {
            return accumulator + 1;
        }


        //窗口闭合的时候，获取计算结果
        //然后将计算结果发送给WindowResult中的process函数
        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }


        @Override
        public Long merge(Long a, Long b) {
            return null;  //只有事件时间的会话窗口才实现，此处不实现只占位用
        }
    }

    //注意输入泛型是Long！！！因为此处输入的是getResult输出的类型
    //ProcessWindowFunction中4个泛型是 <输入，输出，key，窗口>
    public static class WindowResult extends ProcessWindowFunction<Long,UrlCountPerWindow,String,TimeWindow>{
        @Override
        public void process(String key, Context context, Iterable<Long> elements, Collector<UrlCountPerWindow> out) throws Exception {
            //迭代器中只有一个元素，就是getResult的返回值
            String url = key;
            Long count =elements.iterator().next();  //去除迭代器中的唯一元素
            Long windowStart = context.window().getStart();
            Long windowEnd = context.window().getEnd();
            out.collect(new UrlCountPerWindow(url,count,windowStart,windowEnd));

        }
    }


}






